package completedRDT;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

/**
 * Reducer for CRDT training
 * @author tigerzhong
 *
 */
public class CRDTTrainReducer extends MapReduceBase implements
	Reducer<Text, Text, Text, Text> {
    /**
     * Number of class columns 
     */
    private int numClass = 1;
    /**
     * Output Value
     */
    private Text valueText = new Text();

    @Override
    public void reduce(Text key, Iterator<Text> values,
	    OutputCollector<Text, Text> context, Reporter reporter)
	    throws IOException {
	double val[] = new double[numClass];		//Class information at one leaf node
	int num = 0;					//Number of instances at one leaf node
	while (values.hasNext()) {
	    /*Split and summary*/
	    String label[] = values.next().toString().split(",", -1);
	    for (int i = 0; i < label.length; i++)
		if (!label[i].equals(""))
		    val[i] += Double.parseDouble(label[i]);
	    num ++;
	}
	StringBuilder res = new StringBuilder();
	for (int i = 0; i < val.length; i++) {
	    res.append(val[i]);
	    res.append(',');
	}
	res.append(num);
	valueText.set(res.toString());
	/* Output format:
	 *	key <Tab> [sum of each class value],[number of instances]
	 */
	context.collect(key, valueText);
    }
    
    @Override
    public void configure(JobConf job) {
	numClass = job.getInt("Data.numClasses", 1);
    }
}
